package com.shujia.streaming

import org.apache.spark.sql.streaming.OutputMode
import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}

object Demo05StructuredStreaming {
  def main(args: Array[String]): Unit = {

    // 创建SparkSession 从Spark 2.0引入的新入口
    val spark: SparkSession = SparkSession
      .builder()
      .appName("Demo05StructuredStreaming")
      .master("local[2]")
      .config("spark.sql.shuffle.partitions", "2")
      .getOrCreate()
    import spark.implicits._
    import org.apache.spark.sql.functions._

    // 读取流的数据
    /**
     * StructuredStreaming实际上还是微批处理
     * 每个批次的延时为100ms 所以看起来好像是来一条数据处理一条
     * 默认会保存状态
     */
    val socketDF: DataFrame = spark
      .readStream
      .format("socket")
      .option("host", "master")
      .option("port", "8888")
      .load()


    // DSL
    socketDF
      .select(explode(split($"value", ",")) as "word")
      .groupBy($"word")
      .agg(count("*") as "cnt")
      .writeStream

      /**
       * OutputMode指定流输出的方式：
       * Append 追加 ： 只适用于一些简单的查询，不能应用在有聚合操作的流中
       * Complete 完整 ：只适用于聚合操作之后的输出，会将所有的结果都输出
       * Update 更新 ：只会将有变化的行的数据输出，不管有无聚合操作都可以使用，如果流中没有聚合操作则等价于Append模式
       */
      .outputMode(OutputMode.Update())
      .format("console")
    //      .start()
    //      .awaitTermination()


    // SQL
    socketDF.createOrReplaceTempView("words")

    spark
      .sql(
        """
          |select t1.word
          |       ,count(*) as cnt
          |from (
          | select explode(split(value,',')) as word
          | from words
          |) t1 group by t1.word
          |""".stripMargin)
      .writeStream
      .outputMode(OutputMode.Complete())
      // 将每个批次的数据转换成DataFrame进行处理
      .foreachBatch((df,l)=>{
        df.write
          .format("jdbc")
          .option("url", "jdbc:mysql://master:3306/bukong?useSSL=false")
          .option("dbtable", "word_cnt")
          .option("user", "root")
          .option("password", "123456")
          .mode(SaveMode.Overwrite) // 会覆盖原有的表 并且表结构会发生变化
          .save()
      })
      .start()
      .awaitTermination()

  }

}
